package me.hl.zk;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * author: planck
 * createTime: 2016/8/23.
 */
public class ZkTest {


    /**
     * CreateMode:
     * PERSISTENT (持续的，相对于EPHEMERAL，不会随着client的断开而消失)
     * PERSISTENT_SEQUENTIAL（持久的且带顺序的）
     * EPHEMERAL (短暂的，生命周期依赖于client session)
     * EPHEMERAL_SEQUENTIAL  (短暂的，带顺序的)
     * <p>
     * Ids:
     * OPEN_ACL_UNSAFE ：完全开放。 事实上这里是采用了world验证模式，由于每个zk连接都有world验证模式.
     * CREATOR_ALL_ACL ：给创建该znode连接所有权限。 事实上这里是采用了auth验证模式，使用sessionID做验证。
     * 所以设置了 CREATOR_ALL_ACL 时，创建该znode的连接可以对该znode做任何修改。
     * READ_ACL_UNSAFE ：所有的客户端都可读。 事实上这里是采用了world验证模式，由于每个zk连接都有world验证模式，
     * 所以znode在设置了READ_ACL_UNSAFE时，所有的连接都可以读该znode。
     */
    private static Logger logger = LoggerFactory.getLogger(ZkTest.class);

    private ZooKeeper zk;

    private boolean isLeader = false;

    private Watcher leaderWatcher;

    private Watcher lockWacher;

    @Test
    public void testCollection() throws Exception {


        zk = new ZooKeeper("172.16.57.19:4180", 10000, x -> logger.info("已经触发 {}", x.getWrapper()));

        List<String> childs = zk.getChildren("/", true);
        childs.forEach((x) -> logger.info("child = {}", x));

        if (!childs.contains("testRootPath")) {
            zk.create("/testRootPath", "testRootData".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            zk.create("/testRootPath/plankTest", "heihei".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        logger.info("{}", new String(zk.getData("/testRootPath", false, null)));

        logger.info("{}", zk.getChildren("/testRootPath", true));

        zk.setData("/testRootPath/plankTest", "hahahahahahaha".getBytes(), -1);

        logger.info("{}", new String(zk.getData("/testRootPath/plankTest", true, null)));


        //删除整个子目录   -1代表version版本号，-1是删除所有版本
        zk.delete("/testRootPath/plankTest", -1);
        zk.delete("/testRootPath", -1);

    }


    public void getLeader() throws Exception {

        Stat stat = zk.exists("/hl/leader", leaderWatcher);
        if (stat != null) {
            following();
        } else {
            while (zk.exists("/hl/leader", null) == null) {
                createLeaderNode();
            }
            if (isLeader) {
                leadering();
            } else {
                following();
            }
        }
    }

    private void createLeaderNode() {
        try {
            zk.create("/hl/leader", "leader".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            isLeader = true;
            logger.info("i'm leader");
        } catch (Exception e) {
            logger.error("can not create leader");
        }
    }

    private void leadering() {
        logger.info("i'm leadering");
    }

    private void following() throws Exception {
        logger.info("i'm following");
    }

    @Test
    public void testLeader() throws Exception {

        zk = getZookeeper();
        Stat stat = zk.exists("/hl", false);
        if (stat == null) {
            zk.create("/hl", "hl".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        leaderWatcher = (x) -> {
            try {
                logger.info("leaderWatcher = {}", leaderWatcher);
                if (zk.exists("/hl/leader", leaderWatcher) == null) {
                    createLeaderNode();
                }
            } catch (Exception e) {
                logger.error("", e);
            }

        };

        getLeader();
        Thread.sleep(10000);
        zk.close();
    }

    private ZooKeeper getZookeeper() throws Exception {
        return new ZooKeeper("172.16.57.19:4180", 3000, x -> logger.info("已经触发 {}", x.getWrapper()));
    }


    public void testLock() throws Exception {
        ZooKeeper zooKeeper = getZookeeper();
        //if ()
    }


}
